package org.apache.flink.runtime.rest.legacy.metrics;

import static java.util.Collections.unmodifiableMap;
import static java.util.Collections.unmodifiableSet;
import static org.apache.flink.runtime.metrics.dump.MetricDump.METRIC_CATEGORY_COUNTER;
import static org.apache.flink.runtime.metrics.dump.MetricDump.METRIC_CATEGORY_GAUGE;
import static org.apache.flink.runtime.metrics.dump.MetricDump.METRIC_CATEGORY_HISTOGRAM;
import static org.apache.flink.runtime.metrics.dump.MetricDump.METRIC_CATEGORY_METER;
import static org.apache.flink.runtime.metrics.dump.QueryScopeInfo.INFO_CATEGORY_CONSUMER;
import static org.apache.flink.runtime.metrics.dump.QueryScopeInfo.INFO_CATEGORY_CONSUMER_MANAGER;
import static org.apache.flink.runtime.metrics.dump.QueryScopeInfo.INFO_CATEGORY_RESOUCE_PROVIDER;
import static org.apache.flink.runtime.metrics.dump.QueryScopeInfo.INFO_CATEGORY_RESOURCESLOT;
import static org.apache.flink.util.Preconditions.checkNotNull;

import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

import javax.annotation.concurrent.ThreadSafe;

import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.runtime.metrics.dump.MetricDump;
import org.apache.flink.runtime.metrics.dump.QueryScopeInfo;
import org.apache.flink.runtime.rest.legacy.metrics.MetricStore.ComponentMetricStore;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/** Nested data-structure to store metrics. */
@ThreadSafe
public class MetricStore {
    private static final Logger LOG = LoggerFactory.getLogger(MetricStore.class);

    /** metrics存储*/
    private final ComponentMetricStore consumerManager = new ComponentMetricStore();
    private final Map<String, ConsumerMetricStore> consumers = new ConcurrentHashMap<>();
    private final Map<String, ResourceProviderMetricStore> resourceProviders = new ConcurrentHashMap<>();

    /**
     * The map holds the attempt number of the representing execution for each subtask of each
     * vertex. The keys and values are JobID -> slotID -> 
     * CurrentExecutionAttemptNumber. When a metric of an execution attempt is added, the metric can
     * also be added to the SubtaskMetricStore when it is of the representing execution.
     */
    private final Map<String, Map<String, Integer>> 
    					representativeAttempts = new ConcurrentHashMap<>();

    /**
     * Remove inactive task managers.
     *
     * @param activeTaskManagers to retain.
     */
    synchronized void retainTaskManagers(List<String> activeTaskManagers) {
        resourceProviders.keySet().retainAll(activeTaskManagers);
    }

    /**
     * Remove inactive jobs..
     *
     * @param activeJobs to retain.
     */
    synchronized void retainJobs(List<String> activeJobs) {
        consumers.keySet().retainAll(activeJobs);
        representativeAttempts.keySet().retainAll(activeJobs);
    }

    public Map<String, Map<String, Integer>> getRepresentativeAttempts() {
        return representativeAttempts;
    }

    /**
     * Add metric dumps to the store.
     *
     * @param metricDumps to add.
     */
    synchronized void addAll(List<MetricDump> metricDumps) {
        for (MetricDump metric : metricDumps) {
            add(metric);
        }
    }

    // -----------------------------------------------------------------------------------------------------------------
    // Accessors for sub MetricStores
    // -----------------------------------------------------------------------------------------------------------------

    /**
     * Returns the {@link ComponentMetricStore} for the JobManager.
     *
     * @return ComponentMetricStore for the JobManager
     */
    public synchronized ComponentMetricStore getConsumerManagerMetricStore() {
        return ComponentMetricStore.unmodifiable(consumerManager);
    }

    /**
     * Returns the {@link ResourceProviderMetricStore} for the given taskmanager ID.
     *
     * @param tmID taskmanager ID
     * @return TaskManagerMetricStore for the given ID, or null if no store for the given argument
     *     exists
     */
    public synchronized ResourceProviderMetricStore getTaskManagerMetricStore(String tmID) {
        return tmID == null ? null : ResourceProviderMetricStore.unmodifiable(resourceProviders.get(tmID));
    }

    /**
     * Returns the {@link ComponentMetricStore} for the given job ID.
     *
     * @param consumerID job ID
     * @return ComponentMetricStore for the given ID, or null if no store for the given argument
     *     exists
     */
    public synchronized ComponentMetricStore getConsumerMetricStore(String consumerID) {
        return consumerID == null ? null : ComponentMetricStore.unmodifiable(consumers.get(consumerID));
    }

    /**
     * Returns the {@link ComponentMetricStore} for the given job/task ID.
     *
     * @param jobID job ID
     * @param taskID task ID
     * @return ComponentMetricStore for given IDs, or null if no store for the given arguments
     *     exists
     */
    public synchronized ResourceSlotMetricStore getTaskMetricStore(String jobID, String taskID) {
        ConsumerMetricStore job = jobID == null ? null : consumers.get(jobID);
        if (job == null || taskID == null) {
            return null;
        }
        return ResourceSlotMetricStore.unmodifiable(job.getSlotMetricStore(taskID));
    }

    public synchronized Map<String, ConsumerMetricStore> getJobs() {
        return unmodifiableMap(consumers);
    }

    public synchronized Map<String, ResourceProviderMetricStore> getTaskManagers() {
        return unmodifiableMap(resourceProviders);
    }

    // 分组件类处理
    @VisibleForTesting
    public void add(MetricDump metric) {
        try {
            QueryScopeInfo info = metric.scopeInfo;
            ResourceProviderMetricStore rp;
            ConsumerMetricStore consumer;
            ResourceSlotMetricStore slot;
            ComponentMetricStore attempt;
            boolean isRepresentativeAttempt;
            //
            String name = info.scope.isEmpty() ? metric.name : info.scope + "." + metric.name;
            if (name.isEmpty()) { // malformed transmission
                return;
            }
            switch (info.getCategory()) {
                case INFO_CATEGORY_CONSUMER_MANAGER:  // 消费者管理器
                    addMetric(consumerManager.metrics, name, metric);
                    break;
                case INFO_CATEGORY_RESOUCE_PROVIDER: // 资源提供者
                    String tmID = ((QueryScopeInfo.ResourceProviderQueryScopeInfo) info).resourceProviderID;
                    rp = resourceProviders.computeIfAbsent(tmID, k -> new ResourceProviderMetricStore());
                    if (name.contains("GarbageCollector")) {
                        String gcName =
                                name.substring(
                                        "Status.JVM.GarbageCollector.".length(),
                                        name.lastIndexOf('.'));
                        rp.addGarbageCollectorName(gcName);
                    }
                    addMetric(rp.metrics, name, metric);
                    break;
                case INFO_CATEGORY_CONSUMER: // 消费者
                    QueryScopeInfo.ConsumerQueryScopeInfo jobInfo =
                            (QueryScopeInfo.ConsumerQueryScopeInfo) info;
                    consumer = consumers.computeIfAbsent(jobInfo.consumerID, k -> new ConsumerMetricStore());
                    addMetric(consumer.metrics, name, metric);
                    break;
                case INFO_CATEGORY_RESOURCESLOT: // 资源槽请求
                    QueryScopeInfo.ResourceSlotQueryScopeInfo slotInfo =
                            		(QueryScopeInfo.ResourceSlotQueryScopeInfo) info;
                    consumer = consumers.computeIfAbsent(slotInfo.providerID, k -> new ConsumerMetricStore());
                    slot = consumer.sltos.computeIfAbsent(slotInfo.slotID, k -> new ResourceSlotMetricStore());

                    // The attempt is the representative one if the current execution attempt
                    // number for the subtask is not present in the currentExecutionAttempts,
                    // which means there should be only one execution
                    isRepresentativeAttempt =
                            isRepresentativeAttempt(
                                    slotInfo.providerID,
                                    slotInfo.slotID,
                                    slotInfo.attemptNumber);

                    // If the attempt is representative one, its metrics can be updated to the
                    // subtask and task metric store.
                    if (isRepresentativeAttempt) {
                        /**
                         * The duplication is intended. Metrics scoped by subtask are useful for
                         * several job/task handlers, while the WebInterface task metric queries
                         * currently do not account for subtasks, so we don't divide by subtask and
                         * instead use the concatenation of subtask index and metric name as the
                         * name for those.
                         */

                    }
                    break;
                default:
                    LOG.debug("Invalid metric dump category: " + info.getCategory());
            }
        } catch (Exception e) {
            LOG.debug("Malformed metric dump.", e);
        }
    }

    // Returns whether the attempt is the representative one. It's also true if the current
    // execution attempt number for the subtask is not present in the currentExecutionAttempts,
    // which means there should be only one execution
    private boolean isRepresentativeAttempt(
            String jobID, String vertexID, int attemptNumber) {
        return Optional.of(representativeAttempts)
                        .map(m -> m.get(jobID))
                        .map(m -> m.get(vertexID))
                        .orElse(attemptNumber)
                == attemptNumber;
    }

    // 
    private void addMetric(Map<String, String> target, String name, MetricDump metric) {
        switch (metric.getCategory()) {
            case METRIC_CATEGORY_COUNTER:
                MetricDump.CounterDump counter = (MetricDump.CounterDump) metric;
                target.put(name, String.valueOf(counter.count));
                break;
            case METRIC_CATEGORY_GAUGE:
                MetricDump.GaugeDump gauge = (MetricDump.GaugeDump) metric;
                target.put(name, gauge.value);
                break;
            case METRIC_CATEGORY_HISTOGRAM:
                MetricDump.HistogramDump histogram = (MetricDump.HistogramDump) metric;
                target.put(name + "_min", String.valueOf(histogram.min));
                target.put(name + "_max", String.valueOf(histogram.max));
                target.put(name + "_mean", String.valueOf(histogram.mean));
                target.put(name + "_median", String.valueOf(histogram.median));
                target.put(name + "_stddev", String.valueOf(histogram.stddev));
                target.put(name + "_p75", String.valueOf(histogram.p75));
                target.put(name + "_p90", String.valueOf(histogram.p90));
                target.put(name + "_p95", String.valueOf(histogram.p95));
                target.put(name + "_p98", String.valueOf(histogram.p98));
                target.put(name + "_p99", String.valueOf(histogram.p99));
                target.put(name + "_p999", String.valueOf(histogram.p999));
                break;
            case METRIC_CATEGORY_METER:
                MetricDump.MeterDump meter = (MetricDump.MeterDump) metric;
                target.put(name, String.valueOf(meter.rate));
                break;
        }
    }

    // -----------------------------------------------------------------------------------------------------------------
    // sub MetricStore classes
    // -----------------------------------------------------------------------------------------------------------------

    /** Structure containing metrics of a single component. */
    @ThreadSafe
    public static class ComponentMetricStore {
        public final Map<String, String> metrics;

        private ComponentMetricStore() {
            this(new ConcurrentHashMap<>());
        }

        private ComponentMetricStore(Map<String, String> metrics) {
            this.metrics = checkNotNull(metrics);
        }

        public String getMetric(String name) {
            return this.metrics.get(name);
        }

        public String getMetric(String name, String defaultValue) {
            String value = this.metrics.get(name);
            return value != null ? value : defaultValue;
        }

        private static ComponentMetricStore unmodifiable(ComponentMetricStore source) {
            if (source == null) {
                return null;
            }
            return new ComponentMetricStore(unmodifiableMap(source.metrics));
        }
    }

    /** Sub-structure containing metrics of a single Resource Provider. */
    @ThreadSafe
    public static class ResourceProviderMetricStore extends ComponentMetricStore {
        public final Set<String> garbageCollectorNames;

        private ResourceProviderMetricStore() {
            this(new ConcurrentHashMap<>(), ConcurrentHashMap.newKeySet());
        }

        private ResourceProviderMetricStore(
                Map<String, String> metrics, Set<String> garbageCollectorNames) {
            super(metrics);
            this.garbageCollectorNames = checkNotNull(garbageCollectorNames);
        }

        private void addGarbageCollectorName(String name) {
            garbageCollectorNames.add(name);
        }

        private static ResourceProviderMetricStore unmodifiable(ResourceProviderMetricStore source) {
            if (source == null) {
                return null;
            }
            return new ResourceProviderMetricStore(
                    unmodifiableMap(source.metrics), unmodifiableSet(source.garbageCollectorNames));
        }
    }

    /** Sub-structure containing metrics of a single Consumer. */
    @ThreadSafe
    private static class ConsumerMetricStore extends ComponentMetricStore {
        private final Map<String, ResourceSlotMetricStore> sltos = new ConcurrentHashMap<>();

        public ResourceSlotMetricStore getSlotMetricStore(String taskID) {
            return taskID == null ? null : sltos.get(taskID);
        }
    }

    /** Sub-structure containing metrics of a resource slot. */
    @ThreadSafe
    public static class ResourceSlotMetricStore extends ComponentMetricStore {

        private ResourceSlotMetricStore() {
            this(new ConcurrentHashMap<>());
        }

        private ResourceSlotMetricStore(
                Map<String, String> metrics) {
            super(metrics);

        }
        
        private static ResourceSlotMetricStore unmodifiable(ResourceSlotMetricStore source) {
            if (source == null) {
                return null;
            }
            return new ResourceSlotMetricStore(
                    unmodifiableMap(source.metrics));
        }
    }

}
